package service

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	pay_pb "bmember/api/pb/client/pay/api/pay"

	pb "bmember/api/pb/server"

	"bmember/pkg/util"

	// "bmember/pkg/mail"
	"gitee.com/quantum-force/of-utils/pkg/mail"

	"bmember/internal/dao"
	"bmember/internal/model"

	"bmember/pkg/lueerror"
)

type schSrv struct {
	*Service
	name string
}

var sSrv *schSrv

// InitScheduleSrv  初始化定时任务相关逻辑层定义
func InitScheduleSrv() {
	sSrv = &schSrv{
		Service: service,
		name:    "group_service",
	}
}

// GetScheduleSrv 获得定时任务相关逻辑层定义
func GetScheduleSrv() *schSrv {
	return sSrv
}

func (srv *schSrv) newCtx(ctx context.Context) context.Context {
	rID := ctx.Value("_requestID")
	return context.WithValue(context.TODO(), "_requestID", rID)
}

func (srv *schSrv) scheduleMail(ctx context.Context, content string) {
	mailMsg := &mail.Message{
		Subject:   "BMember 定时任务异常通知",
		Content:   bytes.NewBufferString(content),
		To:        []string{"backend@luedongtech.com"},
		Extension: nil,
	}
	_ = mail.AsyncSend(mailMsg, func(err error) {
		if err != nil {
			srv.logger.For(ctx).Error(fmt.Sprintf("邮件发送错误: %v", err))
		}
	})
}

func (srv *schSrv) checkRechargeStatus(ctx context.Context, cChan <-chan int64, sch *model.Schedules) {
	srv.logger.For(ctx).Info(fmt.Sprintf("run id %d", sch.Id))
	defer func() {
		// 持久化定时任务状态
		_ = dao.GetScheduleDao().UpdateScheduleById(ctx, sch)
		<-cChan
		if rec := recover(); rec != nil {
			srv.scheduleMail(ctx, fmt.Sprintf("panic 异常 %v", rec))
		}
	}()
	// 定时任务参数
	param := sch.GetRechargeParam()
	if param == nil {
		sch.Failed(errors.New("参数解析异常"))
		return
	}
	_, err := dao.Transaction(ctx, func(ctx context.Context) (interface{}, error) {
		order, err := dao.GetOrdersDao().LockRechargeOrderById(ctx, param.OrderId)
		if err != nil {
			sch.AddRetry(err)
			return nil, lueerror.DbErr(err)
		}
		// 订单不存在
		if order == nil {
			sch.Failed(errors.New("订单不存在"))
			return nil, lueerror.OrderNotExist(nil)
		}
		// 定时是否已经在终态，返回成功
		if order.IsFinished() {
			sch.Success()
			return nil, nil
		}
		// 查询订单状态
		payOrder, err := dao.GetOrdersDao().GetPayOrderDetail(ctx, order.PayOrderNo)
		if err != nil {
			sch.AddRetry(err)
			return nil, err
		}
		// 更新订单状态
		err = order.UpdatePayStatus(payOrder.ToRechargeOrderStatus(), payOrder.UserPayed, payOrder.PayOrderNo, payOrder.PayTime)
		if err != nil {
			sch.AddRetry(err)
			return nil, err
		}
		err = dao.GetOrdersDao().UpdateRechargeOrderById(ctx, order)
		if err != nil {
			sch.AddRetry(err)
			return nil, err
		}
		// 订单再终态，标记为成功
		if order.IsFinished() {
			sch.Success()
		} else {
			sch.AddRetry(nil)
		}
		return nil, nil
	})
	if err != nil {
		srv.scheduleMail(ctx, err.Error())
		srv.logger.For(ctx).Error(fmt.Sprintf("recharge pay sechdule errror %s", err.Error()))
	} else if sch.NeedMail() {
		srv.scheduleMail(ctx, fmt.Sprintf("重试次数超过 %d", sch.Retry))
	}
}

// RechargeOrderPayCheck 充值订单，支付检查
func (srv *schSrv) RechargeOrderPayCheck(ctx context.Context, limit int) lueerror.LueError {
	if limit == 0 {
		limit = 10
	}
	// 获取定时任务，并锁定
	schs, err := dao.GetScheduleDao().LoadWaitingSchedules(ctx, model.ScheduleRechargeOrder, limit)
	if err != nil {
		return lueerror.DbErr(err)
	}

	ctx = srv.newCtx(ctx)
	hChan := make(chan int64, 2)
	for _, sch := range schs {
		hChan <- sch.Id
		go srv.checkRechargeStatus(ctx, hChan, sch)
	}
	return nil
}

// CreateSchedule 创建定时任务
func (srv *schSrv) CreateSchedule(ctx context.Context, schType int8, param []byte) (int64, lueerror.LueError) {
	sch := model.NewSechdule(schType, param, util.GetNow().Add(1*time.Second))
	var err error
	sch.Id, err = dao.GetScheduleDao().CreateSchedule(ctx, sch)
	if err != nil {
		return 0, lueerror.DbErr(err)
	}
	return sch.Id, nil
}

// 重试消息
func (srv *schSrv) redoMessage(ctx context.Context, msgId int64) lueerror.LueError {
	_, err := dao.Transaction(ctx, func(ctx context.Context) (interface{}, error) {
		msg, err := dao.GetMessageDao().LockMessageById(ctx, msgId)
		if err != nil {
			return nil, lueerror.DbErr(err)
		}
		if msg == nil {
			return nil, nil
		}
		if !msg.IsForConsuming() {
			return nil, nil
		}
		if msg.IsFinished() {
			return nil, nil
		}
		var errL lueerror.LueError
		switch msg.Topic {
		case srv.cfg.Topics.TopicMemberRegister:
			payload := &pb.MemberRegisterNotify{}
			if err = json.Unmarshal(msg.Content, payload); err != nil {
				errL = lueerror.InvalidRequest(err)
			} else {
				errL = GetMemberSrv().MemberRegisterNotify(ctx, payload.BrandId, payload.MemberId)
			}
		case srv.cfg.Topics.TopicCardSend:
			payload := &pb.CardSendingNotify{}
			if err = json.Unmarshal(msg.Content, payload); err != nil {
				errL = lueerror.InvalidRequest(err)
			} else {
				errL = GetMemberSrv().CardSendingNotify(ctx, payload.BrandId, payload.AccId)
			}
		case srv.cfg.Topics.TopicOrderPayFinish:
			payload := &pay_pb.PayOrderView{}
			if err = json.Unmarshal(msg.Content, payload); err != nil {
				errL = lueerror.InvalidRequest(err)
			} else {
				errL = GetAssetSrv().PayOrderFinishNotify(ctx, payload)
			}
		case srv.cfg.Topics.TopicOrderRefundFinish:
			payload := &pay_pb.RefundOrderView{}
			if err = json.Unmarshal(msg.Content, payload); err != nil {
				errL = lueerror.InvalidRequest(err)
			} else {
				errL = GetAssetSrv().RefundOrderFinishNotify(ctx, payload)
			}
		default:
			return nil, nil
		}
		if errL != nil {
			// 失败，需要重试
			msg.NeedRetry()
		} else {
			// 成功
			msg.Success()
		}
		err = dao.GetMessageDao().UpdateMessageById(ctx, msg)
		if err != nil {
			return nil, lueerror.DbErr(err)
		}
		return nil, nil
	})
	return parseLueerror(err)
}

// RedoMessages 重新消费消息
func (srv *schSrv) RedoMessages(ctx context.Context, msgIds []int64) lueerror.LueError {
	if len(msgIds) == 0 {
		return nil
	}
	for _, msgId := range msgIds {
		errL := srv.redoMessage(ctx, msgId)
		if errL != nil {
			return errL
		}
	}
	return nil
}
